use crate::core::config::Conf;
use tokio::stream::StreamExt;
use tokio_util::codec::{BytesCodec, Decoder};
use tokio::runtime::Runtime;
use tokio::net::TcpListener;
use futures::sink::SinkExt;
use bytes::{Bytes};
use crate::core::kfk;
use log::{info,error,trace};
use crate::core::stat::KEY;
use crate::core::stat::Status;
use std::str;
use crossbeam_channel::{unbounded,Sender};
//use std::sync::mpsc::{Sender, channel};
use bytes::BytesMut;
use rand::Rng;
use byteorder::{ByteOrder,LittleEndian};

pub fn run(){
    let c = Conf::get_config(None);
    let mut runtime = Runtime::new().unwrap();
    //生成通信管道
    let thread_num = c.brokers.len();
    let mut kfk_msg_receivers = vec![];
    let mut senders = vec![];
    for _ in 0..thread_num {
        let (tx, rx) = unbounded::<BytesMut>();
        //let (tx, rx) = channel::<BytesMut>();
        kfk_msg_receivers.push(rx);
        senders.push(tx);
    }
    //启动状态线程
    let status_sender = Status::start_status_server();
    //启动kafka的producer服务群
    kfk::init_producer_group(thread_num,status_sender,kfk_msg_receivers);
    //启动接收客户端消息
    match c.stype.as_str() {
        "udp" => udp_server(),
            _ => runtime.block_on(tcp_server(thread_num,&senders))
    }
}

pub async fn tcp_server(thread_num : usize, senders : &[Sender<BytesMut>]) {
    let c = Conf::get_config(None);
    let server_addr = format!("{}:{}",c.addr,c.port);
    let mut listener = match TcpListener::bind(server_addr).await {
        Ok(listener) => {
            info!("proxy tcp server is running on port : {} ~ ",c.port);
            listener
        },
        _ => {
            error!("the port : {} is in use ~ ",c.port);
            std::process::exit(0);
        }
    };
    loop {
        let (socket, addr) = match listener.accept().await {
            Ok((_socket, _addr)) => (_socket, _addr),
                          Err(e) => {
                                info!("accept error : {:?}",e);
                                continue;
                          },
        };
        let s = senders[rand::thread_rng().gen_range(0,thread_num)].clone();
        trace!("income from {}",addr);
        tokio::spawn(async move {
            let mut framed = BytesCodec::new().framed(socket);
            while let Some(message) = framed.next().await {
                match message {
                    Ok(mut bytes) => {
                        if bytes.len() < 16 {
                            trace!("proxy err msg is : {:?}", str::from_utf8(&bytes[..]));
                            continue;
                        }
                        match str::from_utf8(&bytes[..16]) {
                            Ok(KEY) => {
                                let status  = match Status::show_status() {
                                    Ok(_status) => _status,
                                            _   => String::from("status service error"),
                                };
                                match framed.send(Bytes::from(status)).await {
                                    Ok(_) =>  {},
                                    Err(_) => error!("once status call lose ~"),
                                };
                            },
                            _ => {
                                loop {
                                    if bytes.len() < 2 {
                                        if bytes.len() != 0 {
                                            trace!("loop lenth < 2 , {}",bytes.len());
                                        }
                                        break;
                                    }
                                    let len = bytes.split_to(2);
                                    let len : usize = LittleEndian::read_u16(&len[..]) as usize;
                                    if bytes.len() < len {
                                        trace!("bytes lenth < len : {:?}", str::from_utf8(&bytes[..]));
                                        break;
                                    }
                                    match s.send(bytes.split_to(len)) {
                                        Ok(_) => continue,
                                        Err(e)   => {
                                            trace!("loop send err : {:?}", e);
                                            continue;
                                        }
                                    };
                                }
                            },
                        }
                    },
                    Err(err) => error!("Socket closed with error: {:?}", err),
                }
            }
        });
    }
}

pub fn udp_server(){
    println!("udp server is running ~");
}